package com.gis.aoitest.state;

import com.gis.aoitest.beans.SensorReading;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.kafka.common.metrics.Sensor;

/**
 * @author LnnuUser
 * @create 2021-09-08-下午8:53
 * 如果温度跳变超过10.0度，输出告警
 */
public class StateTest03_KeyedStateApplicationCase {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // socket文本流
        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 9999);

        // 转换成SensorReading类型
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        } );

        // 定义一个flatmap操作，检测温度跳变，输出报警
        SingleOutputStreamOperator<Tuple3<String, Double, Double>> resultMap = dataStream.keyBy("id")
                .flatMap(new TempChangeWarning(10.0));

        resultMap.print();


        env.execute();
    }

    public static class TempChangeWarningFlatMap extends RichFlatMapFunction<SensorReading, Tuple3<String, Double, Double>>{

        @Override
        public void open(Configuration parameters) throws Exception {

        }

        @Override
        public void flatMap(SensorReading sensorReading, Collector<Tuple3<String, Double, Double>> collector) throws Exception {

        }
    }

    // 自定义函数类
    public static class TempChangeWarning extends RichFlatMapFunction<SensorReading, Tuple3<String, Double, Double>> {

        // 私有属性，当前温度跳变的阈值
        private Double threshold;

        public TempChangeWarning(Double threshold) {
            this.threshold = threshold;
        }

        // 定义状态，保存上一次的温度值
        private ValueState<Double> lastTempState;

        @Override
        public void open(Configuration parameters) throws Exception {
            lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("last-temp", Double.class));

        }

        @Override
        public void flatMap(SensorReading value, Collector<Tuple3<String, Double, Double>> out) throws Exception {

            // 获取上一次的温度值状态
            Double lastTemp = lastTempState.value();

            // 如果状态不为null，就判断两次差值
            if (lastTemp != null) {
                double diff = Math.abs(value.getTemperature() - lastTemp);
                if (diff >= threshold)
                    out.collect(new Tuple3<>(value.getId(), lastTemp, value.getTemperature()));

            }

            // 更新状态
            lastTempState.update(value.getTemperature());
        }

        @Override
        public void close() throws Exception {
            lastTempState.clear();
        }
    }
}
